package org.jeecg.modules.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;

@Service
@Slf4j
public class MqttSubscriberService {

    @Autowired
    private MqttPahoClientFactory mqttClientFactory;

    @Autowired
    private MessageChannel mqttInputChannel;

    @Autowired
    private MessageChannel mqttOutboundChannel;

    @Value("${mqtt.client-id}")
    private String clientId;

    @Value("${mqtt.topic-prefix:ftts}")
    private String topicPrefix;

    private final Map<String, MqttPahoMessageDrivenChannelAdapter> adapters =
            new HashMap<>();

    /**
     * 应用启动时自动订阅主题
     */
    @PostConstruct
    public void init() {
        // 订阅以topicPrefix开头的所有主题
        subscribe(topicPrefix + "/#", 1);
    }

    /**
     * 订阅指定主题
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        // 如果已经订阅了该主题，则先取消订阅
        if (adapters.containsKey(topic)) {
            unsubscribe(topic);
        }

        // 创建新的适配器
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(
                        clientId + "_subscriber_" + topic.replace("/", "_"),
                        mqttClientFactory,
                        topic
                );

        adapter.setQos(qos);
        adapter.setOutputChannel(mqttInputChannel);
        adapter.start();

        // 保存适配器引用，以便后续取消订阅
        adapters.put(topic, adapter);
        log.info("订阅指定主题: {}", topic);
    }

    /**
     * 取消订阅指定主题
     * @param topic
     */
    public void unsubscribe(String topic) {
        MqttPahoMessageDrivenChannelAdapter adapter = adapters.get(topic);
        if (adapter != null) {
            adapter.stop();
            adapters.remove(topic);
            log.info("取消订阅指定主题: {}", topic);
        }
    }

    /**
     * 发送消息方法
     * @param topicSuffix
     * @param message
     */
    public void sendMessage(String topicSuffix, String message) {
        String fullTopic = topicPrefix + topicSuffix;
        org.springframework.messaging.Message<String> mqttMessage = MessageBuilder.withPayload(message)
                .setHeader(MqttHeaders.TOPIC, fullTopic)
                .build();
        mqttOutboundChannel.send(mqttMessage);
        log.info("发送消息至主题: {}, 消息: {}", fullTopic, message);
    }

    /**
     * 发送消息方法（可指定QoS）
     * @param topicSuffix
     * @param message
     * @param qos
     */
    public void sendMessage(String topicSuffix, String message, int qos) {
        String fullTopic = topicPrefix + topicSuffix;
        org.springframework.messaging.Message<String> mqttMessage = MessageBuilder.withPayload(message)
                .setHeader(MqttHeaders.TOPIC, fullTopic)
                .setHeader(MqttHeaders.QOS, qos)
                .build();
        mqttOutboundChannel.send(mqttMessage);
        log.info("发送消息至主题: {}, QoS: {}, 消息: {}", fullTopic, qos, message);
    }

}

